import java.io.File

import Tool.isNX
import org.apache.commons.lang3.StringUtils
import scopt.OParser

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
import scala.io.Source


/**
  * Created by yz on 2019/1/23
  */


case class Config(snpFile: File = new File("G:\\rice90k_data_deal\\test\\Kb_5K_raw_Gt.hmp.txt.add_90k_name.hmp_10_f4.txt"),
                  regionFile: File = new File("G:\\rice90k_data_deal\\test\\region.txt"),
                  hybrid: Double = 0.5, rowRate: Double = 0.5, minSnp: Int = 5, sameRate: Double = 0.8,
                  groupNxRate: Double = 0.5, threadNum: Int = 2, outFile: File = new File("G:\\rice90k_data_deal\\test\\out.txt")
                 )

object Main {

  case class Node(sample: String, values: Seq[String])


  def main(args: Array[String]): Unit = {

    val builder = OParser.builder[Config]
    val parser1 = {
      import builder._
      //s r h w m e n t o
      OParser.sequence(
        programName("scopt"),
        head("scopt", "4.x"),
        opt[File]('s', "snp").valueName("<file>").action { (x, c) =>
          c.copy(snpFile = x)
        }.text("please input snp data file!"),
        opt[File]('r', "region").valueName("<file>").action { (x, c) =>
          c.copy(regionFile = x)
        }.text("please input region info file!"),
        opt[File]('o', "out").valueName("<file>").action { (x, c) =>
          c.copy(outFile = x)
        }.text("please input out file!"),
        opt[Double]('h', "hybrid").valueName("<double>").action { (x, c) =>
          c.copy(hybrid = x)
        }.text("snp NX(NN or hybrid) rate,must < this(column)!"),
        opt[Double]('w', "row").valueName("<double>").action { (x, c) =>
          c.copy(rowRate = x)
        }.text("snp NX(NN or hybrid) rate,must < this(row)!"),
        opt[Int]('m', "minSnp").valueName("<int>").action { (x, c) =>
          c.copy(minSnp = x)
        }.text("every region snp num, must  >= this!"),
        opt[Double]('e', "same").valueName("<double>").action { (x, c) =>
          c.copy(sameRate = x)
        }.text("node1 and node2 same standard, must  >= this!"),
        opt[Double]('n', "group-rate").valueName("<double>").action { (x, c) =>
          c.copy(groupNxRate = x)
        }.text("every region snp NX(NN or hybrid) rate, must  < this!"),
        opt[Int]('t', "thread").valueName("<int>").action { (x, c) =>
          c.copy(threadNum = x)
        }.text("thread num!"),
        help("help").text("prints this usage text")
      )
    }
    OParser.parse(parser1, args, Config()) match {
      case Some(config) => exec(config)
      case _ =>
    }

    def exec(config: Config) = {
      val hybrid = config.hybrid
      val rowRate = config.rowRate
      val minSnp = config.minSnp
      val sameRate = config.sameRate
      val groupNxRate = config.groupNxRate
      val startTime = System.currentTimeMillis()
      val file = config.snpFile
      val regionFile = config.regionFile
      import Implicits.MyFile
      import Implicits.RegionLines
      val regionDatas = regionFile.lines.regionDatas
      val outLines = ArrayBuffer[String]()
      println(Utils.getTime(startTime))
      val source = Source.fromFile(file)
      val linesIter = source.getLines()
      import Implicits.SnpLines1
      val linesStream = linesIter.toStream
      val filterLines = linesStream.nxfiy.filterEmptyPos.filterRow(rowRate)
      val (idInfos,sampleGenesMap) = filterLines.getNeedInfos
      Utils.logTime(startTime)
      val filterSampleGenesMap = Tool.columnFilter(sampleGenesMap, hybrid)
      Utils.logTime(startTime)

      import Implicits.MyLines1
      val headers = linesStream.headers
      outLines += (ArrayBuffer("Chr", "Start", "End", "Genotype", "group") ++= headers.drop(3)).mkString("\t")

      val finalThreadNum = config.threadNum
      val map = regionDatas.zipWithIndex.map { case (v, i) =>
        val j = (i % finalThreadNum) + 1
        (j, v)
      }.groupBy(_._1).mapValues(_.map(_._2))
      val f = map.map { case (i, tmpRegionDatas) =>
        Future {
          var j = 0
          tmpRegionDatas.map { regionData =>
            val needIdInfos = idInfos.filter { x =>
              regionData.chr == x.chr && x.pos >= regionData.start && x.pos <= regionData.end
            }
            j += 1
            if (j % 10 == 0) {
              println("--time--")
              Utils.logTime(startTime)
            }

            if (needIdInfos.size >= minSnp) {
              val nodes = Tool.getNodes(filterSampleGenesMap, needIdInfos)
              val filterNodes = Tool.filterNodes(nodes, groupNxRate)
              val map = Tool.getClusterMap(filterNodes, sameRate)
              val groupMap = Tool.getGroupMapV2(map)
              val representMap = Tool.getRepresentMap(groupMap)
              val sampleGroupMap = getSampleGroup(groupMap)
              (ArrayBuffer(regionData.chr.toString, regionData.start.toString, regionData.end.
                toString, representMap.values.mkString(","), representMap.keySet.mkString(",")) ++=
                headers.drop(3).map { x =>
                  sampleGroupMap.getOrElse(x, "-")
                }).mkString("\t")
            } else ""
          }.filter(StringUtils.isNotBlank(_))
        }
      }.reduceLeft((x, y) =>
        x zip y map { t =>
          t._1 ++ t._2
        })
      outLines ++= Utils.execFuture(f).sortBy { line =>
        val columns = line.split("\t")
        (columns(0).toInt, columns(1).toInt)
      }
      Utils.lines2File(config.outFile, outLines)
      Utils.logTime(startTime)
      source.close()

    }


  }

  def getSampleGroup(map: mutable.LinkedHashMap[String, ArrayBuffer[Node]]) = {
    map.flatMap { case (key, nodes) =>
      nodes.map { node =>
        (node.sample, key)
      }
    }

  }


}
